package com.example.demo.service.impl;

import com.example.demo.PrepareJdbc;
import com.example.demo.dto.DataxDto;
import com.example.demo.dto.SyncSqlDto;
import com.example.demo.service.DataxToL1Service;
import com.example.demo.service.SyncSql;
import com.example.demo.threadpool.ThreadPoolManager;
import com.example.demo.util.StreamGroupByUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.apm.toolkit.trace.TraceCrossThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * ClassName:DataxToL1ServiceImpl
 * Package:com.example.demo.service.impl
 * TODO:
 *
 * @date:2021/8/26 16:28
 * @author:Vicky(陈烟星)
 */
@Service
public class DataxToL1ServiceImpl implements DataxToL1Service, DisposableBean {

    private static Logger logger = LoggerFactory.getLogger(DataxToL1ServiceImpl.class);

    private boolean shutdown;

    @Autowired
    private ThreadPoolManager threadPoolManager;
    @Autowired
    private PrepareJdbc prepareJdbc;
    @Autowired
    private SyncSql syncSql;

    @Override
    public void dataxToL1(DataxDto dataxDto) {
        if (shutdown) {
            logger.error("系统关闭中,拒绝处理同步数据请求");
            return;
        }
        boolean exitData = dataxDto != null && StringUtils.isNotBlank(dataxDto.getSourceTableName()) && StringUtils.isNotBlank(dataxDto.getTargetTableName());
        List<SyncSqlDto> syncSqlDtoList = new ArrayList<>();
        if (exitData) {
            SyncSqlDto syncSqlDto = syncSql.querySyncSql(dataxDto.getSourceTableName(), dataxDto.getTargetTableName());
            if (syncSqlDto != null) {
                syncSqlDtoList.add(syncSqlDto);
            }
        } else {
            syncSqlDtoList = syncSql.queryDealAllSyncSql(null);
        }
        if (CollectionUtils.isEmpty(syncSqlDtoList)) {
            logger.info("没有数据同步sql配置");
            return;
        }
        LogPunchRunnable logPunchRunnable = null;

        Map<Integer, List<SyncSqlDto>> syncSqlDtoMap = syncSqlDtoList.stream()
                .collect(StreamGroupByUtil.groupByWithNullKeys(SyncSqlDto::getDealPrio));

        for (Map.Entry<Integer, List<SyncSqlDto>> m : syncSqlDtoMap.entrySet()) {
            List<SyncSqlDto> syncSqlDtos = m.getValue();
            if (CollectionUtils.isEmpty(syncSqlDtos)) {
                continue;
            }
            for (SyncSqlDto syncSqlDto : syncSqlDtos) {
                //可以自定义一些参数
                logPunchRunnable = new LogPunchRunnable();
                logPunchRunnable.prepareJdbc = prepareJdbc;
                logPunchRunnable.syncSqlDto = syncSqlDto;
                //提交线程任务
                submitThreadTask(logPunchRunnable);
            }
        }

    }

    private void submitThreadTask(LogPunchRunnable logPunchRunnable) {
        ThreadPoolManager.ThreadPoolExecutorWrapper threadPoolExecutorWrapper = threadPoolManager.getDataThreadPool();
        for (; ; ) {
            if (threadPoolExecutorWrapper.isFree()) {
                threadPoolExecutorWrapper.submit(logPunchRunnable);
            } else {
                logger.debug("log punch pool is not free. waiting...");
            }
        }

    }

    @Override
    public void destroy() throws Exception {
        shutdown = true;
    }

    @TraceCrossThread//支持跨线程的链路追踪
    private static class LogPunchRunnable implements Runnable {
        private PrepareJdbc prepareJdbc;
        private SyncSqlDto syncSqlDto;

        @Override
        public void run() {
            try {
                long threadId = Thread.currentThread().getId();
                String srcTableName = syncSqlDto.getSrcTableName();
                String tgtTableName = syncSqlDto.getTgtTableName();
                long currentTimeMillis = System.currentTimeMillis();
                prepareJdbc.submitPrepareJdbc(syncSqlDto);
                logger.info("线程id:{},srcTableName:{},tgtTableName:{},cost time:{}", threadId, srcTableName, tgtTableName, System.currentTimeMillis() - currentTimeMillis);
            } catch (SQLException e) {
                logger.error("任务执行异常:{}", e.getMessage());
            }
        }
    }


}
